package com.yifeng.repo.controller.rocketmq.message.consume;

import com.talkyun.utils.Looper;
import com.yifeng.repo.base.api.TimeRange;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.consumer.store.ReadOffsetType;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.Collections;
import java.util.List;
import java.util.function.Consumer;

/**
 * Created by daibing on 2023/8/6.
 */
@Slf4j
public class AssignPullConsumerLooper extends Looper {
    private final MessageQueue messageQueue;
    private final DefaultLitePullConsumer pullConsumer;
    private final List<TimeRange> timeRangeList;
    private final Consumer<List<MessageExt>> msgListConsumer;
    private final boolean seekBackWhenError;
    private boolean paused = false;

    public AssignPullConsumerLooper(String name, int interval, int errorInterval, MessageQueue messageQueue,
                                    DefaultLitePullConsumer pullConsumer, List<TimeRange> timeRangeList,
                                    Consumer<List<MessageExt>> msgListConsumer, boolean seekBackWhenError) {
        super(name, interval, errorInterval);
        this.messageQueue = messageQueue;
        this.pullConsumer = pullConsumer;
        this.timeRangeList = timeRangeList;
        this.msgListConsumer = msgListConsumer;
        this.seekBackWhenError = seekBackWhenError;
        log.info("AssignPullConsumerLooper ok: name={}, timeRangeList={}", name, timeRangeList);
    }

    @Override
    protected void loop() throws Throwable {
        // 检查当前时间段
        boolean nowInside = TimeRange.nowInside(timeRangeList);
        if (nowInside) {
            // 1.如果已经暂停，就恢复消费，并且设置状态标记
            if (paused) {
                pullConsumer.resume(Collections.singleton(messageQueue));
                paused = false;
            }

            // 2.拉取&消费消息
            List<MessageExt> messages = pullConsumer.poll();
            msgListConsumer.accept(messages);
            messages.clear();

            // 3.对应构造时将自动提交位点属性设置为false，这里将消费位点提交到内存中的OffsetStore，最终通过定时任务将消费位点提交给broker
            pullConsumer.commitSync();
        } else {
            // 1.如果已经暂停，继续睡眠1分钟
            if (paused) {
                doSleep(60 * 1000);
                return;
            }
            // 2.如果没有暂停，就暂停，并且设置状态标记，以及睡眠1分钟
            pullConsumer.pause(Collections.singleton(messageQueue));
            paused = true;
            doSleep(60 * 1000);
        }
    }

    @Override
    protected void loopThrowable(Throwable t) {
        if (seekBackWhenError) {
            long offset = pullConsumer.getOffsetStore().readOffset(messageQueue, ReadOffsetType.READ_FROM_MEMORY);
            try {
                pullConsumer.seek(messageQueue, offset);
            } catch (MQClientException e) {
                log.warn("consumer {} seek {} failed: error=", messageQueue.getTopic() + "-" + messageQueue.getQueueId(), offset, t);
            }
        }
        log.warn("consumer {} failed: error=", messageQueue.getTopic() + "-" + messageQueue.getQueueId(), t);
    }

    @Override
    protected void clear() {
        pullConsumer.unsubscribe(messageQueue.getTopic());
        pullConsumer.shutdown();
    }

    public DefaultLitePullConsumer getPullConsumer() {
        return pullConsumer;
    }
}
